
/* Author: Manfred Gschweidl */

#include "GlobusGramJobsCallback.h"

using namespace globus_job;

namespace globus_gram_jobs {

//callback funtion registered by callback server
static void callback_func(void * user_callback_arg,
                   char * job_contact,
                   int state,
                   int errorcode)
{
    GlobusGramJobsCallback* Monitor = (GlobusGramJobsCallback*) user_callback_arg;
    GlobusGramJob* job = Monitor->getJob(job_contact);
    
    switch(state)
    {
	    case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_IN:
			cout << "Staging file in on: " << job_contact << endl;
			break;
	    case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_OUT:
			cout << "Staging file out  on: " << job_contact << endl;
			break;
	    case GLOBUS_GRAM_PROTOCOL_JOB_STATE_PENDING:
			break; /* Reports state change to the user */
	
	    case GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE:
			break; /* Reports state change to the user */
	   
	    case GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED:
			Monitor->remove(job_contact);
			job->setFailed();
			job->setDone();
			cerr << "Job Failed on: " << job_contact << endl;
			break; /* Reports state change to the user */
	
	    case GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE:
			cout << "Job Finished on: " << job_contact << endl;
			cout << "GlobusGramJob hjobcontact: " << job->getJobContact() << endl;
			Monitor->remove(job_contact);
			job->setDone();
			break; /* Reports state change to the user */
    }
}

//callback funtion registered by job for receiving jobcontact
static void request_callback(void * user_callback_arg,
                             globus_gram_protocol_error_t failure_code,
                             const char * job_contact,
                             globus_gram_protocol_job_state_t state,
                             globus_gram_protocol_error_t errorcode) {
    GlobusGramJob*  Request = (GlobusGramJob*) user_callback_arg;
    cout << "Contact on the server " << job_contact << endl;

    if (failure_code==0) {
    	Request->setJobContact(job_contact);
    	if(Request->onlyGetJobHandle()) {
    		Request->setDone();
    	}
    }
    else {
       	cout << "Error during the code submission" << endl << "Error Code:" << failure_code << endl;
		Request->setDone();
		Request->setFailed();
    }
}

}

namespace globus_job {

GlobusGramJob::GlobusGramJob(GlobusGramJobsCallback* f) : failed(false),jobcontact(NULL),callback(f),
	only_get_jobHandle(false) {
	globus_mutex_init(&jobcontactmutex,GLOBUS_NULL);
}

GlobusGramJob::GlobusGramJob(GlobusGramJobsCallback* f, string jc) : failed(false),callback(f),
	only_get_jobHandle(false)
{
	//need to copy string because globus resuses memroy allocted by pointer
	jobcontact = new char[strlen(jc.c_str())+1];
	char* temp=strcpy( jobcontact, jc.c_str());
	globus_mutex_init(&jobcontactmutex,GLOBUS_NULL);
}

GlobusGramJob::GlobusGramJob(string jc) : failed(false),callback(NULL),
	only_get_jobHandle(false)
{
	//need to copy string because globus resuses memroy allocted by pointer
	jobcontact = new char[strlen(jc.c_str())+1];
	char* temp=strcpy( jobcontact, jc.c_str());
	globus_mutex_init(&jobcontactmutex,GLOBUS_NULL);
}

GlobusGramJob::~GlobusGramJob() {
	if (jobcontact!=NULL) {
		globus_gram_client_job_contact_free(jobcontact);
		jobcontact=NULL;
	}
	globus_mutex_destroy(&jobcontactmutex);
}

//get job status from globus - blocking implementation
int GlobusGramJob::getJobStatus() {
	if(jobcontact == NULL) {
		cout << "no jobcontact to query status" << endl;
		return -1;
	}
	int state = 0;
	int failureCode = 0;
	int rc = globus_gram_client_job_status(jobcontact, &state, &failureCode);
	if(rc != GLOBUS_SUCCESS) {
		cout << "error on quering current job status of job handle: " << jobcontact << endl;
	}
		
    switch(state)
    {
		case GLOBUS_GRAM_PROTOCOL_JOB_STATE_PENDING:
			rc = 1;
			break; /* Reports state change to the user */
		case GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE:
			rc = 2;
			break; /* Reports state change to the user */
    	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED:
    		if(callback != NULL) {
				callback->remove(jobcontact);
    		}
			setFailed();
			setDone();
			cerr << "Job Failed on: " << jobcontact << endl;
			rc = 4;
			break; /* Reports state change to the user */    	
		case GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE:
			cout << "Job Finished on: " << jobcontact << endl;
			if(callback != NULL) {
				callback->remove(jobcontact);
			}
			rc = 3;
			setDone();
			break; /* Reports state change to the user */
		case GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED:
			rc = 5;
			break; /* Reports state change to the user */
		case GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED:
			rc = 6;
			break; /* Reports state change to the user */
		case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_IN:
			cout << "Staging file in on: " << jobcontact << endl;
			rc = 7;
			break;
		case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_OUT:
			cout << "Staging file out  on: " << jobcontact << endl;
			rc = 8;
			break;
		default:
			cout << "job status is unkown" << endl;
			rc = -1;
    }
    return rc;
}

bool GlobusGramJob::onlyGetJobHandle() {
	return only_get_jobHandle;
}

void GlobusGramJob::setOnlyGetJobHandle(bool state) {
	only_get_jobHandle = state;
}

//submit callback function sets jobcontact
void GlobusGramJob::setJobContact(const char* c) {
	//need to copy again string because of pointer change by globus
	jobcontact = new char[strlen(c)+1];
	char* temp=strcpy( jobcontact, c);
    callback->add(c,this);
}

//submit job to globus and register callback funtion to receive jobcontact
bool GlobusGramJob::submit(string res, string rsl) {
	cout << "Submitting job on " << res << endl;
	failed=false;
	int rc = globus_gram_client_register_job_request(res.c_str(),
			rsl.c_str(),
			GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL,
			callback->getURL(),
			GLOBUS_GRAM_CLIENT_NO_ATTR,
			globus_gram_jobs::request_callback,
			(void*) this);
	if (rc != 0) /* if there is an error */
	{
		printf("gram error: %d - %s\n", 
			rc, 
			/* translate the error into english */
			globus_gram_client_error_string(rc));
		return true;
	} else {
		return false;
	}
}

//possible reuse of object for further job submissions by callback server
void GlobusGramJob::proceed() {
	globus_mutex_lock(&jobcontactmutex);
	/* Free up the resources of the job_contact, as the job is over, and
	 * the contact is now useless.
	 */
	if (jobcontact!=NULL) {
		globus_gram_client_job_contact_free(jobcontact);
		jobcontact=NULL;
	}

	GlobusCallback::proceed();
	globus_mutex_unlock(&jobcontactmutex);
}

//cancle job - blocking implementation
bool GlobusGramJob::cancel() {
	if(jobcontact == NULL) {
		cout << "no jobcontact to query status" << endl;
		return false;
	}
	int rc;
	printf("\tsending cancel to job manager...\n");

	if ((rc = globus_gram_client_job_cancel(jobcontact)) != GLOBUS_SUCCESS)
	{
		printf("\tFailed to cancel job.\n");
		printf("\tgram error: %d - %s\n", 
			rc,
			globus_gram_client_error_string(rc));
		return false;
	} else {
		printf("\tjob cancel was successful.\n");
		return true;
	}
}

void GlobusGramJob::setFailed() {
	failed=true;
}

bool GlobusGramJob::hasFailed() {
	return failed;
}

string GlobusGramJob::getJobContact() {
	return string(jobcontact);
}

GlobusGramJobsCallback::GlobusGramJobsCallback() {
	globus_module_activate(GLOBUS_GRAM_CLIENT_MODULE);
	globus_mutex_init(&JobsTableMutex, GLOBUS_NULL);
	//register callback funtion for job status changes
	globus_gram_client_callback_allow( globus_gram_jobs::callback_func,
		(void *) this,
		&callback_contact);
	//cout << "Gram contact " << callback_contact << endl;
}

char* GlobusGramJobsCallback::getURL() {
	return callback_contact;
}

GlobusGramJob* GlobusGramJobsCallback::getJob(char* s) {
	return JobsTable[s];
}

GlobusGramJobsCallback::~GlobusGramJobsCallback() {
	//cout << callback_contact << " destroyed" << endl;
	//deregister callback function
	globus_gram_client_callback_disallow(callback_contact);
	globus_free(callback_contact);
	globus_mutex_destroy(&JobsTableMutex);
	globus_module_activate(GLOBUS_GRAM_CLIENT_MODULE);
}

//add job to internal map
void GlobusGramJobsCallback::add(string jobcontact,GlobusGramJob* job) {
	lock();
	JobsTable[jobcontact]=job;
	unLock();
}

//remove job form itnernal map identified by jobcontact
void GlobusGramJobsCallback::remove(char* jobcontact){ 
	lock();
	JobsTable.erase(jobcontact);
	unLock();
}

void GlobusGramJobsCallback::lock() {
	globus_mutex_lock(&JobsTableMutex);
}

void GlobusGramJobsCallback::unLock() {
	globus_mutex_unlock(&JobsTableMutex);
}

}
